use bytes::BytesMut;
use log::{debug, error, info};
use rand::Rng;
use std::{fmt::Write, sync::Arc};
use tokio::{
    io::{AsyncReadExt, AsyncWriteExt, BufWriter},
    net::TcpStream,
    sync::{broadcast::Sender, Mutex},
};

use crate::{
    cache::Metadata, wal::segment::WalReader, Argument, Command, Error, Input, SendMessage,
    BYTE_CAPACITY,
};

use super::{
    command_handler::{create_topic_command, delete_topic_command, list_command, produce_command},
    consume_command,
};

#[derive(Debug)]
pub struct Connection {
    pub stream: BufWriter<TcpStream>,
    pub buf: BytesMut,
    pub sender: Sender<SendMessage>,
}

impl Connection {
    pub fn new(socket: TcpStream, sender: Sender<SendMessage>) -> Self {
        Connection {
            stream: BufWriter::new(socket),
            buf: BytesMut::with_capacity(BYTE_CAPACITY),
            sender,
        }
    }

    pub async fn read_socket(&mut self) -> Result<usize, Error> {
        self.buf.clear();
        let len = match self.stream.read_buf(&mut self.buf).await {
            // socket closed
            Ok(n) if n == 0 => {
                self.stream.shutdown().await?;
                return Err(Error::CustomError("socket closed".to_owned()));
            },
            Ok(n) => n,
            Err(e) => {
                error!("failed to read from socket; err = {:?}", e);
                return Err(Error::CustomError("failed to read from socket".to_owned()));
            }
        };
        //info!("read socket: {:?}", String::from_utf8_lossy(&self.buf));
        Ok(len)
    }

    pub async fn read_command(&mut self) -> Result<Input, Error> {
        let len = self.read_socket().await?;
        Ok(Input::new(&self.buf, len))
    }

    pub async fn read_message(&mut self, topic: &str, partition: Option<u16>) -> Result<(), Error> {
        let partition = match partition {
            Some(p) => p,
            None => rand::thread_rng().gen_range(0..3), //todo!() 随机数有待优化
        };

        self.read_socket().await?;
        let len = self.buf.len();
        if len >= 2 && &self.buf[len - 2..len] == b"\r\n" {
            self.buf.truncate(self.buf.len() - 2); //remove \r\n
        }

        let mut msg_buf = BytesMut::with_capacity(self.buf.len());
        msg_buf.extend_from_slice(&self.buf);
        let msg = SendMessage::new(topic.to_string(), partition, msg_buf);
        let _ = self.sender.send(msg);
        Ok(())
    }

    pub async fn write_message(&mut self) -> Result<(), Error> {
        //println!("{:?}", self.buf);
        self.stream.write(&self.buf).await?;
        self.stream.flush().await?;
        self.buf.clear();
        Ok(())
    }

    // pub async fn pull_message(&mut self) -> Result<(), Error> {
    //     self.read_socket().await?;
    //     self.buf
    //     Ok(())
    // }
}

pub struct Handler {
    pub metadata: Arc<Mutex<Metadata>>,
}

impl Handler {
    pub fn new(metadata: Arc<Mutex<Metadata>>) -> Self {
        Handler { metadata }
    }

    pub async fn command(&mut self, mut conn: Connection) -> Result<(), Error> {
        let mut input = conn.read_command().await?;
        let token = match input.analysis() {
            Some(token) => token,
            None => return Err(Error::CustomError("command error".to_owned())),
        };

        //println!("{:?}", token);
        match token.command {
            Command::CreateTopic => {
                create_topic_command(token, conn, Arc::clone(&self.metadata)).await?;
            }
            Command::List => {
                list_command(conn, Arc::clone(&self.metadata)).await?;
            }
            Command::Describe => {}
            Command::Alter => {}
            Command::GetOffset => {}
            Command::DeleteTopic => {
                delete_topic_command(token, conn, Arc::clone(&self.metadata)).await?;
            }
            Command::Produce => {
                produce_command(token, conn, Arc::clone(&self.metadata), &self).await?;
            }
            Command::Consume => {
                consume_command(token, conn, Arc::clone(&self.metadata)).await?;
            }
            Command::DeleteGroup => {}
            Command::Pull => {}
            Command::None => {}
        }

        Ok(())
    }

    pub async fn produce_message<'a>(
        &self,
        topic: &'a str,
        parition: Option<u16>,
        mut conn: Connection,
    ) -> Result<(), Error> {
        loop {
            conn.read_message(topic, parition).await?;
        }
    }
}

#[derive(Debug)]
pub struct Consumer {
    pub topic: String,
    pub partitions: Vec<u16>,
    pub conn: Connection,
}

impl Consumer {
    pub async fn new(topic: String, partitions: Vec<u16>, conn: Connection) -> Self {
        Consumer {
            topic,
            partitions,
            conn,
        }
    }

    pub async fn send_message(&mut self) -> Result<(), Error> {
        let mut wal = WalReader::new();
        loop {
            let mut input = self.conn.read_command().await?;
            let token = match input.analysis() {
                Some(token) => token,
                None => return Err(Error::CustomError("command error".to_owned())),
            };
            //info!("{:?}", token);

            if Command::Pull != token.command {
                return Err(Error::CustomError("command error".to_owned()));
            }

            let mut pull_topic = None;
            let mut partition_vec = Vec::<u16>::new();
            let mut pull_len = 10;
            for arg in token.args {
                match arg {
                    Argument::Topic { topic } => {
                        pull_topic = Some(topic);
                    }
                    Argument::Partitions { partitions } => {
                        partition_vec.push(partitions);
                    }
                    Argument::Len { len } => {
                        pull_len = len;
                    }
                    _ => {}
                }
            }

            if partition_vec.is_empty() {
                partition_vec.extend(&self.partitions);
            }

            match wal
                .read_message(
                    &pull_topic.get_or_insert(self.topic.clone()),
                    &partition_vec,
                    pull_len,
                )
                .await
            {
                Ok(msgs) => {
                    //println!("{:#?}", msgs);
                    // if 0 == msgs.len() {
                    //     sleep(Duration::from_millis(200)).await;
                    // }
                    self.conn.buf.clear();
                    for msg in msgs {
                        debug!("{:#?}", msg);
                        let _ = self.conn.buf.write_str(&format!("{:?}\n", msg));
                        self.conn.write_message().await?;
                    }
                }
                Err(_) => {
                    println!("error");
                    //sleep(Duration::from_millis(200)).await;
                }
            };
        }
    }
}
